package com.mkz.producer.feign;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import com.mkz.core.common.util.ResponseResult;
import com.mkz.producer.feign.ProducerClient;
import com.mkz.producer.service.MsgLogService;

import lombok.AllArgsConstructor;

/**
 * ClassName: ProducerController 
 * @Description: Kafka生产者实现类
 * @author: KuangJianyu
 * @date: 2020年4月16日
 */
@RestController
@AllArgsConstructor
public class IProducerClient implements ProducerClient {
    private final  MsgLogService msgLogService;
    private final KafkaTemplate<String, String> kafkaTemplate;

	@Override
	@PutMapping("/basicSend")
	public ResponseResult<Boolean> basicSend(@RequestParam String topic, @RequestParam String content, @RequestParam String bustype) {
		ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, content);
		boolean sendok = true;
		try {
			kafkaTemplate.send(record);
		} catch (Exception e) {
			e.printStackTrace();
			sendok = false;
		}
		if(sendok) {//发送成功
			msgLogService.addSendMsg(topic, content, bustype, "1");
		}else {//发送失败
			msgLogService.addSendMsg(topic, content, bustype, "0");
		}
		return ResponseResult.ok(sendok);
	}
}